package com.vlin.api.mq.utils;

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;

import com.vlin.api.mq.config.MqConfigs;
import com.vlin.api.mq.config.MqEnv;

public class PulsarConsumerPoolFactory {

	 public static ConcurrentHashMap<Integer, Consumer> getConsumerPool() {
	        return InitializingConsumerPool.pool;
	    }


	    private static class InitializingConsumerPool {
	        private static final ConcurrentHashMap<Integer, Consumer> pool = new ConcurrentHashMap<Integer, Consumer>() {{
	            for (int i = 1; i <= MqConfigs.CONSUMER_NUM; i++) {
	                //
	                int retryTimes = getRetryTimes();
	                Consumer consumer = getConsumerInstance();
	                while (Objects.isNull(consumer) && retryTimes > 0) {
	                    retryTimes--;
	                    consumer = getConsumerInstance();
	                }
	                //
	                if (Objects.isNull(consumer)) {
	                    System.out.println("PulsarConsumerPoolFactory create consumer error! and have retry two times!");
	                    continue;
	                }
	                put(i, consumer);
	            }
	        }};


	        private static Consumer getConsumerInstance() {
	            try {
	                return PulsarClientFactory.getPulsarClient().newConsumer().topic(String.format("%s/out/%s", MqConfigs.accessId, MqEnv.PROD.getValue()))
	                        .subscriptionName(String.format("%s-sub", MqConfigs.accessId)).subscriptionType(SubscriptionType.Failover)
	                        .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(MqConfigs.MAX_REDELIVER_COUNT).build()).subscribe();
	            } catch (PulsarClientException e) {
	                System.err.println("PulsarConsumerPoolFactory init error! e=" + e);
	                return null;
	            }
	        }


	        public static int getRetryTimes() {
	            return 2;
	        }
	    }
}
